import json
import time
from datetime import datetime
from kafka import KafkaProducer
from kafka.admin import KafkaAdminClient, NewTopic
from kafka.errors import TopicAlreadyExistsError

# 创建Kafka生产者
producer = KafkaProducer(
    bootstrap_servers=['127.0.0.1:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    api_version=(0, 10, 1)
)

# 发送测试数据
def send_matched_data():
    current_time = datetime.now()
    timestamp_str = current_time.isoformat()
    timestamp_ms = int(current_time.timestamp() * 1000)
    sequence_id = int(timestamp_ms % 1000000)  # 使用时间戳的后6位作为整数ID
    
    # 真实数据 - 完整格式匹配TrueData模型
    true_data = {
        "timestamp": timestamp_str,
        "load": 100.0,
        "max_temp": 25.5,
        "min_temp": 18.2,
        "avg_temp": 21.8,
        "humidity": 65.0,
        "rainfall": 0.0,
        "data_source": "test",
        "sequence_id": sequence_id
    }
    
    # 预测数据 - 误差较大
    predict_data = {
        "timestamp": timestamp_str,
        "predicted_load": 150.0,  # 50%误差
        "request_id": f"req_{timestamp_ms}",
        "model_info": {
            "model_name": "lstm_model",
            "version": "1.0"
        },
        "sequence_id": sequence_id
    }
    
    # 发送数据
    producer.send('true_data', true_data)
    producer.send('predict_data', predict_data)
    producer.flush()
    
    print(f"发送数据 - 时间戳: {timestamp_str}, 真实负荷: {true_data['load']}, 预测负荷: {predict_data['predicted_load']}, 误差: {abs(true_data['load'] - predict_data['predicted_load'])}")
    return timestamp_str

print("开始发送连续5次误差较大的数据...")
timestamps = []
for i in range(5):
    ts = send_matched_data()
    timestamps.append(ts)
    time.sleep(3)  # 等待3秒

print(f"\n已发送5次误差较大的数据，时间戳: {timestamps}")
print("等待CEP处理...")
time.sleep(5)

# 发送误差较小的数据
print("\n发送误差较小的数据...")
for i in range(3):
    current_time = datetime.now()
    timestamp_str = current_time.isoformat()
    timestamp_ms = int(current_time.timestamp() * 1000)
    sequence_id = int(timestamp_ms % 1000000)  # 使用时间戳的后6位作为整数ID
    
    true_data = {
        "timestamp": timestamp_str,
        "load": 100.0,
        "max_temp": 25.5,
        "min_temp": 18.2,
        "avg_temp": 21.8,
        "humidity": 65.0,
        "rainfall": 0.0,
        "data_source": "test",
        "sequence_id": sequence_id
    }
    
    predict_data = {
        "timestamp": timestamp_str,
        "predicted_load": 102.0,  # 2%误差
        "request_id": f"req_{timestamp_ms}",
        "model_info": {
            "model_name": "lstm_model",
            "version": "1.0"
        },
        "sequence_id": sequence_id
    }
    
    producer.send('true_data', true_data)
    producer.send('predict_data', predict_data)
    producer.flush()
    
    print(f"发送数据 - 时间戳: {timestamp_str}, 真实负荷: {true_data['load']}, 预测负荷: {predict_data['predicted_load']}, 误差: {abs(true_data['load'] - predict_data['predicted_load'])}")
    time.sleep(2)

producer.close()
print("\n测试完成！请查看Flink CEP输出。")